package com.eshipenet.shipnet.dp_unpacker

import org.apache.flink.streaming.api.scala._

object Main {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val dpStream = EdgeChannelDataPacketStreamProvider.create(env)

    val unpacked = dpStream
      .map(packet => Unpacker.unpack(packet))
      .filter(_.isSuccess)
      .map(_.get)
      .flatMap(d => d)

    unpacked.print("unpacked")

    unpacked.addSink(EdgeChannelDataSinkProvider.create(env))

    val name = this.getClass.getName
    env.execute(name)
  }

}
